package com.ds.lens.flink.monitor.monitorjob.function.detector;

import com.ds.lens.data.common.constant.MonitorStateEnum;
import com.ds.lens.data.common.document.DataPoint;
import com.ds.lens.data.common.document.dto.MonitorEvent;
import com.ds.lens.flink.monitor.common.bo.MonitorDefinition;
import com.ds.lens.flink.monitor.common.BuilderUtil;
import com.ds.lens.flink.monitor.common.LensCache;
import com.ds.lens.flink.monitor.common.bo.Metric;
import org.springframework.util.CollectionUtils;

import javax.persistence.Transient;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;

public class MonitorEventDetectorChain implements MonitorEventDetector {
    @Transient
    private MonitorDefinition definition;
    private List<MonitorEventDetector> detectors = new ArrayList<>();
    @Transient
    private LensCache<String, Metric> detectEventCache = new LensCache<>(null, 10000, 10);
    private volatile boolean sorted = false;

    @Override
    public MonitorStateEnum getMonitorState() {
        throw new UnsupportedOperationException();
    }

    @Override
    public MonitorEvent detect(Metric metric) {
        if (CollectionUtils.isEmpty(metric.getValues())) {
            return generateEvent(metric);
        }
        String key = generateMetricKey(metric);
        try {
            Metric oldMetric = detectEventCache.get(key);
            if (oldMetric != null) {
                for (DataPoint dataPoint : metric.getValues()) {
                    int index = oldMetric.getValues().indexOf(dataPoint);
                    if (index >= 0) {
                        DataPoint oldPoint = oldMetric.getValues().get(index);
                        if (oldPoint.getState() != null) {
                            dataPoint.setState(oldPoint.getState());
                        }
                    }
                }
            }
            return generateEvent(metric);
        } finally {
            detectEventCache.put(key, metric);
        }
    }

    public void addDetector(MonitorEventDetector detector) {
        this.detectors.add(detector);
    }

    @Override
    public void setDefinition(MonitorDefinition definition) {
        this.definition = definition;
        for (MonitorEventDetector detector : detectors) {
            detector.setDefinition(definition);
        }
    }

    private MonitorEvent generateEvent(Metric metric) {
        if (!sorted) {
            Collections.sort(detectors, new MonitorEventDetectorComparator());
            sorted = true;
        }
        MonitorEvent curEvent = null;
        for (MonitorEventDetector detector : detectors) {
            curEvent = detector.detect(metric);
            if (curEvent != null) {
                break;
            }
//            if (curEvent == null || curEvent.getTriggered().getState().getSeverity() < event.getTriggered().getState().getSeverity()) {
//                curEvent = event;
//            }
//            if (curEvent.getTriggered().getState().getSeverity() >= MonitorStateEnum.ALARM_LEVEL_1.getSeverity()) {
//                break;
//            }
        }
        return curEvent != null ? curEvent : BuilderUtil.buildMonitorEvent(definition, metric, null, null, MonitorStateEnum.OK);
    }

    public String generateMetricKey(Metric metric) {
        StringBuilder builder = new StringBuilder()
                .append(metric.getMonitorKey().getMonitorId())
                .append(metric.getMonitorKey().getVersion())
                .append(metric.getMetricName())
                .append(metric.getAggrMethod().name());
        Map<String, Object> map = metric.getGroupBys();
        if (!CollectionUtils.isEmpty(map)) {
            for (String key : map.keySet()) {
                builder.append(key).append(map.get(key));
            }
        }
        return builder.toString();
    }
}
